package cn.doitedu.cn.doitedu.rtdw.feature_etl;

import cn.doitedu.cn.doitedu.rtdw.ddl.CatalogDatabaseSwitch;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/2
 * @Desc: 学大数据，上多易教育
 *    广告点击率预估，特征数据加工
 **/
public class ADClickRatePredict {
    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 切换catalog为hiveCatalog
        CatalogDatabaseSwitch.switchCatalogDatabase("doit40-hive","doit40_rtdw",tenv);

        //tenv.executeSql("select * from dwd_kafka").print();

        // 在行为日志流中匹配模式：  曝光,点击+,加购
        tenv.executeSql(
                " with tmp as (                                                          " +
                        " SELECT                                                                 " +
                        "    user_id,                                                            " +
                        "    event_id,                                                           " +
                        "    event_time,                                                         " +
                        "    properties['ad_id'] as ad_id,                                       " +
                        "    properties['ad_tracking_id'] as ad_tracking_id,                     " +
                        "    rt                                                                  " +
                        " FROM dwd_kafka                                                         " +
                        " )                                                                      " +
                        "                                                                        " +
                        "                                                                        " +
                        " SELECT                                                                 " +
                        "   a_userid as user_id,                                                 " +
                        "    a_adid as ad_id,                                                    " +
                        "    show_time,                                                          " +
                        "    click_time,                                                         " +
                        "    add_cart_time                                                       " +
                        " from tmp                                                               " +
                        " match_recognize (                                                      " +
                        " partition by user_id                                                   " +
                        " order by rt                                                            " +
                        " measures                                                               " +
                        "    A.ad_id as a_adid,                                                  " +
                        "    A.event_time as show_time,                                          " +
                        "    A.user_id as a_userid,                                              " +
                        "    FIRST(B.event_time) as click_time,                                  " +
                        "    C.event_time as add_cart_time                                       " +
                        " one row per match                                                      " +
                        " after match skip to next row                                           " +
                        " pattern (A B+ C) WITHIN INTERVAL '5' MINUTE                            " +
                        " define                                                                 " +
                        "   A as A.event_id = 'ad_show',                                         " +
                        "   B as B.event_id = 'ad_click' AND A.ad_tracking_id=B.ad_tracking_id,  " +
                        "   C as C.event_id = 'add_cart'                                         " +
                        " ) re                                                                   "
        ).print();




    }
}
